/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.opencfg.core.mode.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * Reactor Class is An implementation of the Reactor pattern
 * <p>
 * 
 * @author Opencfg Software Foundation
 * @since 0.0.1-SNAPSHOT
 * @version $Id: Reactor.java 2011-09-06 22:21:53 reymondtu $
 */
public class Reactor implements Runnable {

    /**
     * Creates a new Reactor
     * @param poolSize the number of WorkerThreads to include in the ThreadPool
     * @param port the port to bind the Reactor to
     * @throws IOException if some I/O problems arise during connection
     */
    public Reactor(final String ip, final int port , final ExecutorService mainExecutor, final ExecutorService workerExecutor) throws IOException {
    	this._ip = ip;
    	this._port = port;
    	this._mainExecutor = mainExecutor;
    	this._workerExecutor = workerExecutor;
    }

    /**
     * Main operation of the Reactor:
     * <ul>
     * <li>Uses the <code>Selector.select()</code> method to find new requests from clients
     * <li>For each request in the selection set:
     * If it is <b>acceptable</b>, use the ConnectionAcceptor to accept it,
     * create a new ConnectionReader for it register it to the Selector
     * <li>If it is <b>readable</b>, use the ConnectionReader to read it,
     * extract messages and insert them to the ThreadPool
     * </ul>
     */
    @Override
    public void run() {
        try {
            // Create a non-blocking server socket channel and bind to to the Reactor port
            ServerSocketChannel ssChannel = ServerSocketChannel.open();
            ssChannel.configureBlocking(false);
            ssChannel.socket().bind(new InetSocketAddress(_ip, _port));

            // Create the selector and bind the server socket to it
            Selector selector = Selector.open();
            ssChannel.register(selector, SelectionKey.OP_ACCEPT, new ConnectionAcceptor(selector, ssChannel, _workerExecutor));

            while (_shouldRun) {
                // Wait for an event
                selector.select();

                // Get list of selection keys with pending events
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();

                // Process each key
                while (it.hasNext()) {
                    // Get the selection key
                    SelectionKey selKey = (SelectionKey)it.next();

                    // Remove it from the list to indicate that it is being processed
                    it.remove();

                    // Check if it's a connection request
                    if (selKey.isValid() && selKey.isAcceptable()) {
                        ConnectionAcceptor connectionAcceptor = (ConnectionAcceptor)selKey.attachment();
                        connectionAcceptor.accept();
                    }
                    // Check if a message has been sent
                    if (selKey.isValid() && selKey.isReadable()) {
                        ConnectionReader connectionReader = (ConnectionReader)selKey.attachment();
                        connectionReader.read();
                    }
                }
            }
        } catch (IOException e) {
            //e.printStackTrace(System.err);
            this.shutdown();
        }
        this.shutdown();
    }

    /**
     * Returns the listening port of the Reactor
     * @return the listening port of the Reactor
     */
    public int getPort() {
        return _port;
    }
    
    public String getIp() {
        return _ip;
    }

    /**
     * Stops the Reactor activity, including the Reactor thread and the Worker
     * Threads in the Thread Pool.
     */
    public void shutdown(){
        _shouldRun = false;
        _mainExecutor.shutdown();
        _workerExecutor.shutdown();
    }

    public static void main(String args[]) {
    	if (args.length!=4) {
    		System.err.println("Usage: java Reactor <ip> <port> <main thread pool size> <worker thread pool size> ");
    		System.exit(1);
    	}
    	
        try {
            Reactor reactor = new Reactor(args[0], Integer.parseInt(args[1]), Executors.newFixedThreadPool(1), Executors.newFixedThreadPool(2));
            Thread st = new Thread(reactor);
            st.start();
            System.out.println("Reactor is ready on ip " + reactor.getIp() + " port " + reactor.getPort());
            st.join();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private final String _ip;
    private final int _port;
    private ExecutorService _mainExecutor;
    private ExecutorService _workerExecutor;
    private volatile boolean _shouldRun = true;
    
}
